package org.fastsyncer.manager.framework;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.fastsyncer.common.constant.CommonConstant;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.common.util.ApplicationUtil;
import org.fastsyncer.manager.listener.TaskHandleListener;
import org.fastsyncer.manager.store.DataCenter;
import org.fastsyncer.manager.util.ManagerUtil;
import org.fastsyncer.worker.framework.WorkerFactory;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ManagerFactory:
 * <p>
 * 执行流程如下：1.分析任务配置 2.持久化任务信息 3.计算拆分任务 4.指派任务给worker 5.接受worker执行结果 6.刷新最新进度
 * </p>
 * 
 * @author YangPei
 * @date 2017年5月13日 下午8:06:44
 * @version 1.0.0
 */
public class ManagerFactory implements Manager, TaskHandleListener {

	private static Logger logger = LoggerFactory.getLogger(ManagerFactory.class);

	private static ManagerFactory instance = new ManagerFactory();

	private ManagerFactory() {
	}

	public static ManagerFactory getInstance() {
		return instance;
	}

	// 待执行任务
	private static ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();

	// 正在执行任务
	private static ConcurrentHashMap<String, String> running = new ConcurrentHashMap<String, String>();

	// 支持同时处理同步任务上限数
	private static Integer max = Integer.parseInt(ApplicationUtil.getKey(CommonConstant.MANAGER_SYNC_NUMBER));

	// 当前正在运行同步任务数
	private static Integer run = 0;

	@Override
	public void handle(String mapping) throws Exception {
		// 1.分析任务配置
		MappingTask mappingTask = ManagerUtil.parseMapping(mapping);
		this.handle(mappingTask);
	}

	@Override
	public void handle(MappingTask mappingTask) throws Exception {
		// 1.分析任务配置
		if (mappingTask == null) {
			logger.error("同步任务配置不能为空!");
			return;
		}
		logger.info("收到同步任务!");

		// 2.计算总进度
		TaskDistributor.getInstance().countProcess(mappingTask);

		// 3.持久化任务
		boolean store = DataCenter.getInstance().storeMapping(mappingTask.getName(), mappingTask);
		if (!store) {
			logger.error("持久化任务失败!");
			return;
		}
		logger.info("持久化任务完成!");

		// 4.添加任务到队列中
		this.addTask(mappingTask.getName());
		logger.info("添加到任务队列中\r");
	}

	@Override
	public void handleIncrement(JSONObject t) throws Exception {
		// 这里是数据增量同步处理，这里会接收增删改操作的数据。
		// 其中，针对DB操作，我们可以接收到同步前和同步后的变化，
		// 暂时直接同步。(这里会在1.0.0版本之后实现审计功能。)
		String taskName = (String) t.get("taskName");
		// 1.获取对应的映射
		MappingTask mappingTask = (MappingTask) DataCenter.getInstance().getMapping(taskName);
		if (mappingTask == null) {
			throw new Exception("驱动配置:" + taskName + "已删除或不存在.");
		}
		
		// TODO 数据分发开始
		// 2.增量同步
		JSONObject handle = WorkerFactory.getInstance().handle(mappingTask, t);
		if (handle != null) {
			// 2.1开个线程去存储错误
			new TaskLogHandle(mappingTask.getName(), handle).start();
			int error = handle.getInt("error");
			int count = mappingTask.getError() + error;
			mappingTask.setError(count);
			DataCenter.getInstance().storeMapping(taskName, mappingTask);
		}
		// TODO 数据分发结束
	}

	/**
	 * 添加执行任务
	 * 
	 * @throws Exception
	 */
	private synchronized void addTask(String taskName) throws Exception {
		// 检查执行任务状态
		validateTask(taskName);

		if (queue.isEmpty()) {
			// 排号
			queue.offer(taskName);
			// 触发调度
			new TouchHandle().start();
		} else {
			// 排号
			queue.offer(taskName);
		}
	}

	/**
	 * 检查执行任务状态
	 * 
	 * @throws Exception
	 */
	public void validateTask(String taskName) throws Exception {
		// 检查是否已添加该执行任务
		if (queue.contains(taskName)) {
			throw new IllegalArgumentException("已启动该任务！");
		}

		// 检查是否已在执行
		if (running.get(taskName) != null) {
			throw new IllegalArgumentException("正在运行该任务！");
		}
	}

	/**
	 * 触发调度
	 */
	class TouchHandle extends Thread {

		@Override
		public void run() {
			while (!queue.isEmpty()) {
				try {
					// 强占同步队列
					occupySyncTask();
					// 线程处理同步任务
					String task = queue.poll();
					new TaskHandle(instance, task).start();
					running.put(task, task);
				} catch (Exception e) {
					logger.error("触发任务调度时异常:" + e.getMessage(), e.getClass());
				}
				try {
					// 避免读取任务太快，控制一下速度
					Thread.sleep(300);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}

	}

	@Override
	public void callback(String taskName) {
		// 移除运行任务
		running.remove(taskName);
		// 释放强占列队
		releaseSyncTask();
	}

	/**
	 * 强占同步队列
	 */
	private static synchronized void occupySyncTask() throws Exception {
		if (run + 1 > max) {
			throw new Exception("同步任务较多，请稍后再试！当前已运行：" + run);
		}
		run += 1;
		logger.info("强占同步队列>>还剩可运行数：" + (max - run));
	}

	/**
	 * 释放强占队列
	 */
	private static synchronized void releaseSyncTask() {
		run -= 1;
		logger.info("释放强占队列>>还剩可运行数：" + (max - run));
	}

	/**
	 * 获取正在排队的同步任务
	 * 
	 * @return
	 */
	public static ConcurrentLinkedQueue<String> getQueue() {
		return queue;
	}

	/**
	 * 获取正在运行的同步任务
	 * 
	 * @return
	 */
	public static ConcurrentHashMap<String, String> getRunning() {
		return running;
	}

	/**
	 * 获取正在运行的同步任务
	 * 
	 * @return
	 */
	public static String getRunning(String taskName) {
		return running.get(taskName);
	}

	/**
	 * 移除正在运行的同步任务
	 * 
	 * @return
	 */
	public static boolean removeRunning(String taskName) {
		running.remove(taskName);
		return true;
	}

	/**
	 * 移除正在排队的同步任务
	 * 
	 * @return
	 */
	public static boolean removeQueue(String taskName) {
		queue.remove(taskName);
		return true;
	}

}
